package net.bwie.realtime.jtp.dws.douyin.log.functions;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.dws.douyin.log.bean.DouYinClass;
import net.bwie.realtime.jtp.dws.douyin.log.bean.TradeMetric;
import net.bwie.realtime.jtp.dws.douyin.log.bean.TradePayInfo;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.HashSet;
import java.util.Set;

import static com.ibm.icu.impl.ValidIdentifiers.Datatype.region;

public class DouYinWindowFunction  implements WindowFunction<TradePayInfo,TradeMetric,Long, TimeWindow> {
    @Override
    public void apply(Long anchorId, TimeWindow window, Iterable<TradePayInfo> input, Collector<TradeMetric> out) throws Exception {
        Set<Long> orderIdSet = new HashSet<>();
        Set<Long> userIdSet = new HashSet<>();

        for (TradePayInfo payInfo : input) {
            orderIdSet.add(payInfo.getOrderId());
            userIdSet.add(payInfo.getUserId());
        }

        long start = window.getStart();
        long end = window.getEnd();
        String windowStartTime = DateTimeUtil.convertLongToString(start, "yyyy-MM-dd HH:mm:ss");
        String windowEndTime = DateTimeUtil.convertLongToString(end, "yyyy-MM-dd HH:mm:ss");

        TradeMetric metric = new TradeMetric(anchorId, windowStartTime, windowEndTime, orderIdSet.size(), userIdSet.size());
        out.collect(metric);
    }
}
